import io.reactivex.Observable
import io.reactivex.schedulers.Schedulers
import io.reactivex.schedulers.TestScheduler
import org.junit.Test
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit

/**
 * 连接的两个来来源 测试终止的影响
 *
 *
 * RxJava - 小心 observeOn 的陷阱
 * https://www.jianshu.com/p/238b286ac69c
 */
class ConcatTest {

    /**
     * 结论： first 完成；再开始 second
     */
    @Test
    fun normal() {
        val first = Observable.create<Int> {
            Thread.sleep(10)
            it.onNext(0)
            Thread.sleep(20)
            it.onNext(1)
            it.onComplete()
        }

        val second = Observable.create<Int> {
            it.onNext(10)
            throw RuntimeException("just test")
        }

        val test = Observable.concat(first, second).test()
        test.assertValues(0, 1, 10)         //  没有线程变化则会按 first 完成；再开始 second
        test.assertError(RuntimeException::class.java)

        val down = CountDownLatch(1)

        val test2 = Observable.concat(first, second)
            .subscribeOn(Schedulers.io())
            .doFinally {
                down.countDown()
            }
            .observeOn(Schedulers.trampoline())
            .test()
        down.await()

        test2.assertValues(0, 1, 10)         //  有线程变化也会按 first 完成；再开始 second
        test2.assertError(RuntimeException::class.java)

    }

    /**
     * 第一个进行map操作等; 结论：无影响
     */
    @Test
    fun firstJust() {
        val first = Observable.just(0)
            .map {
                Thread.sleep(1000)
                it+1
            }

        val second = Observable.create<Int> {
            throw RuntimeException("just test")
        }

        val down = CountDownLatch(1)
        val test = Observable.concat(first, second)
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.newThread())
            .doFinally { down.countDown() }
            .test()
        down.await()
        test.assertValues(1)
        test.assertError(RuntimeException::class.java)

    }

    /**
     * 订阅消耗onNext需要时间时, 模拟安卓切到主线程运行需要Loop队列
     * onError 发生时会清空 onNext, 所以只打出1
     */
    @Test
    fun subjectBlock() {
        val first = Observable.just(1, 2)
        val second = Observable.create<Int> {
            throw RuntimeException("just test")
        }

        val down = CountDownLatch(1)
        Observable.concat(first, second)
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.newThread()) // .observeOn(Schedulers.newThread() ，true) 会打印全部 1, 2
            .doFinally { down.countDown() }
            .subscribe({
                Thread.sleep(200)
                println(it)
            }, ::println)


        down.await()

    }

}